package com.wyw.kafka.learning.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.wyw.kafka.learning.vo.WywMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @Title   kafka消费者
 * @Description
 * @Author Mr Wu yewen.wu.china@gmail.com
 * Update History:
 * Author        Time            Content
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaConsumer {
    private final ObjectMapper mapper;

    /**
     * 监听kafka消息并消费
     * @param record    消息
     */
    @KafkaListener(topics = {"wyw-springboot"}, groupId = "wyw-springboot-kafka")
    public void listener01(ConsumerRecord<String, String> record) throws Exception {
        String key = record.key();
        String value = record.value();
        WywMessage wywMessage = mapper.readValue(value, WywMessage.class);

        log.info("in listener01 consume kafka message: {}, {}",
                key, mapper.writeValueAsString(wywMessage));
    }

    /**
     * 监听 Kafka 消息并消费
     * */
    @KafkaListener(topics = {"wyw-springboot"}, groupId = "wyw-springboot-kafka-1")
    public void listener02(ConsumerRecord<?, ?> record) throws Exception {

        Optional<?> _kafkaMessage = Optional.ofNullable(record.value());
        if (_kafkaMessage.isPresent()) {
            Object message = _kafkaMessage.get();
            WywMessage kafkaMessage = mapper.readValue(message.toString(),
                    WywMessage.class);//不知道类型记得try catch包装
            log.info("in listener02 consume kafka message: [{}]",
                    mapper.writeValueAsString(kafkaMessage));
        }
    }
}
